// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <sstream>
#include <string>

using namespace std;

#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>

#include <xbase_utility.hpp>
#include <naming_strategy.hpp>

#include "io_service_pool_impl.hpp"
#include "session_manager.hpp"
#include "tablet_scanner.hpp"

using namespace xbase;

typedef boost::system::error_code error_code;

typedef ::kb::network_byteorder byteorder;
typedef ::kb::binary_iarchive<byteorder> iarchive;

void on_get_next_row(tablet_scanner::ptr_type ts_ptr, row_result::ptr_type row_ptr, const error_code &ec)
{
	if(ec)
	{
		std::cout << "get next row failed : " << ec
			<< " - " << ec.message() << std::endl;
	}
	else
	{
		cout << "row: {row_name: " << dump_string_hex(row_ptr->m_row_name) << " columns: {";

		row_result::cell_vector::iterator iter = row_ptr->m_cell_vector.begin();
		while (iter != row_ptr->m_cell_vector.end())
		{
			cell &c = *iter;
			unsigned version_count = c.get_version_count();

			std::cout << "{column_name: " << c.m_column_name;
			std::cout << " version_count: " << version_count << " versions: {";

			unsigned show_version_count = min(unsigned(3), version_count);

			for (unsigned i=0;i<show_version_count;i++)
			{
				cell::version v;
				c.get_version(i, v);
				cout << "{" << v.first << "," << dump_string_hex(v.second) << "}, ";
			}
			cout << "}} ";

			++iter;
		}
		cout << "}}" << endl;

		std::cout << "pause...";
		std::cin.get();

		ts_ptr->async_get_next_row(boost::bind(on_get_next_row, ts_ptr, _1, _2));
	}
}

int main(int argc, char **argv)
{
	if(argc<5)
	{
		cout <<argv[0]<<" tablet_ip tablet_port table_name tablet_id [column_names]"<< endl;
		return 0;
	}

	boost::asio::ip::tcp::endpoint endpoint(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);

	string table_name(argv[3]);
	uint64_t tablet_id = atoll(argv[4]);
	string start_row;
	string end_row;
	vector<string> column_names;
	
	if (argc >= 6)
	{
		for (int i=5;i<argc;i++)
		{
			column_names.push_back(argv[i]);
		}
	}

	//
	// 首先初始化service_pool并启动
	// 

	boost::shared_ptr<io_service_pool> pool(new impl::single_io_service());
	client_options options;
	pool->set_options(options);
	pool->run();

	session_manager sm(*pool);

	cout<<"tablet_id: "<<tablet_id<<" table_name: "<<table_name<<" start_row: "<<start_row<<" end_row: "<<end_row<<endl;

	string tablet_name = naming_strategy::tablet_name(tablet_id, table_name, start_row, end_row);

	cout<<"tablet_name: "<<tablet_name<<endl;

	tablet_scan_info si(tablet_name, start_row, end_row);
	si.m_column_names = column_names;

	tablet_scanner::ptr_type ts_ptr(new tablet_scanner(*pool, endpoint, sm, si));

	ts_ptr->async_get_next_row(boost::bind(on_get_next_row, ts_ptr, _1, _2));

	// 结束
	std::string input;
	while(std::cin >> input)
	{
		if(input == "close")
		{
			sm.close();
		}
		else if(input == "quit")
		{
			break;
		}
	}

	pool->stop();
	pool->join();

	return 0;
}
